package com.bs.flink.sink;

import com.alibaba.fastjson.JSONObject;
import com.bs.flink.vo.TableData;
import lombok.extern.log4j.Log4j2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.springframework.boot.Banner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.context.ApplicationContext;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;


/**
 * @author bingshao
 * @date 2022/11/7
 **/
@EnableAutoConfiguration
@Log4j2
public class RedisSink extends RichSinkFunction<String> {

    private RedisTemplate<String, Object> redisTemplate;

    private ApplicationContext applicationContext;


    @Override
    public void open(Configuration parameters) throws Exception {
        log.info("redisOpen=========");
        SpringApplication application = new SpringApplication(RedisSink.class);
        application.setBannerMode(Banner.Mode.OFF);
        this.applicationContext = application.run(new String[]{});
        this.redisTemplate = redisTemplateSerializer((RedisTemplate) applicationContext.getBean("redisTemplate"));
    }

    @Override
    public void close() throws Exception {
        super.close();
    }

    @Override
    public void invoke(String value, Context context) throws Exception {
        TableData tableData = JSONObject.parseObject(value, TableData.class);
        String type = tableData.getType();
        log.info("redisData=========" + value);
        if ("delete".equals(type)) {
            redisTemplate.opsForHash().delete(tableData.getDatabase(), JSONObject.parseObject(tableData.getBefore().toString()).get("id").toString());
        } else {
            redisTemplate.opsForHash().put(tableData.getDatabase(), JSONObject.parseObject(tableData.getAfter().toString()).get("id").toString(), tableData.getAfter().toString());
        }
    }

    public RedisTemplate<String, Object> redisTemplateSerializer(RedisTemplate<String, Object> redisTemplate) {
        redisTemplate.setKeySerializer(new StringRedisSerializer());
        redisTemplate.setHashKeySerializer(new StringRedisSerializer());
        redisTemplate.setHashValueSerializer(new StringRedisSerializer());
        redisTemplate.setValueSerializer(new StringRedisSerializer());
        return redisTemplate;
    }
}
